

#include "master_slave.h"
// #define _GNU_SOURCE // 编译时链接了
#include <string.h>
#include <unistd.h>
#include <sys/sendfile.h>
#include <fcntl.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <arpa/inet.h>

#if MASTER_SLAVE
#define MAX_RETRIES 10  // 最大重试次数
#define SLEEP_US 100000 // 每次重试休眠时间（微秒，100ms）

MasterSlaveConfig g_ms_config = {0};

// ======================= 辅助函数 =================================
// 64位字节序转换
static inline uint64_t htonll(uint64_t host64)
{
    if (__BYTE_ORDER__ == __ORDER_BIG_ENDIAN__)
    {
        return host64;
    }
    else
    {
        return ((uint64_t)htonl(host64 & 0xFFFFFFFF) << 32) | htonl(host64 >> 32);
    }
}

static inline uint64_t ntohll(uint64_t net64)
{
    return htonll(net64);
}

// 用rand()生成16位十六进制主机ID（例如：a3f2c7d8b1e09576）
void generate_master_id(char *id)
{
    // 生成8个16位随机数，拼接为16位十六进制字符串
    for (int i = 0; i < MASTER_ID_LEN; i += 2)
    {
        uint16_t rand_val = rand() % 0x10000; // 0~65535
        sprintf(&id[i], "%04x", rand_val);    // 格式化为4位十六进制（补0）
    }
    id[MASTER_ID_LEN] = '\0'; // 结束符
}

SlaveMeta read_slave_meta()
{
    SlaveMeta meta = {0};
    meta.aof_offset = 0;
    FILE *fp = fopen(META_FILE, "rb");
    if (fp)
    {
        fread(&meta, sizeof(meta), 1, fp);
        fclose(fp);
    }
    return meta;
}

void write_slave_meta(const SlaveMeta *meta)
{
    FILE *fp = fopen(META_FILE, "wb");
    if (fp)
    {
        fwrite(meta, sizeof(*meta), 1, fp);
        fclose(fp);
    }
}

int recv_nolock(int fd, void *buf, size_t len, int flags)
{
    int is_noblock = is_non_blocking(fd);

    if (is_noblock == -1)
    {
        return -1;
    }
    if (is_noblock == 0)
    {
        // 如果是阻塞的设置成非阻塞的
        set_socket_noblock(fd);
    }

    ssize_t total = 0;
    int retry_count = 0; // 尝试读取的次数限制
    while (total != len)
    {
        int n = recv(fd, buf + total, len - total, flags);

        if (n > 0)
        {
            // 读取到数据：累计长度，重置重试计数（避免因中间短暂无数据而误判超时）
            total += n;
            retry_count = 0; // 读到数据后，重试计数清零
            continue;
        }

        if (n == 0)
        {
            // 对方关闭连接（或到达EOF），无法继续读取
            // fprintf(stderr, "Connection closed by peer\n");
            return 0;
            break;
        }

        // n == -1：读取失败，判断错误类型
        if (errno == EINTR)
        {
            // 被信号中断：无需计数，直接重试（非数据问题）
            continue;
        }
        else if (errno == EAGAIN || errno == EWOULDBLOCK)
        {
            // 暂时无数据：休眠后重试，累计次数
            retry_count++;
            usleep(SLEEP_US); // 休眠指定时间（微秒）
            if (retry_count >= MAX_RETRIES)
            {
                // LOG("Timeout: max retries (%d) recv_nolock\n", MAX_RETRIES);
                return -1;
            }
        }
        else
        {
            // 其他错误（如文件描述符无效等）
            return -1;
        }
    }

    if (is_noblock == 0)
    {
        // 如果之前是阻塞的设置回去
        set_socket_blocking(fd);
    }

    return total;
}

int send_nolock(int fd, void *buf, size_t len, int flags)
{
    int is_noblock = is_non_blocking(fd);

    if (is_noblock == -1)
    {
        return -1;
    }
    if (is_noblock == 0)
    {
        // 如果是阻塞的设置成非阻塞的
        set_socket_noblock(fd);
    }

    ssize_t total = 0;
    int retry_count = 0; // 尝试读取的次数限制
    while (total != len)
    {
        int n = send(fd, buf + total, len - total, flags);

        if (n > 0)
        {
            // 读取到数据：累计长度，重置重试计数（避免因中间短暂无数据而误判超时）
            total += n;
            retry_count = 0; // 读到数据后，重试计数清零
            continue;
        }

        if (n == 0)
        {
            // 对方关闭连接（或到达EOF），无法继续读取
            // fprintf(stderr, "Connection closed by peer\n");
            return 0;
            break;
        }

        // n == -1：读取失败，判断错误类型
        if (errno == EINTR)
        {
            // 被信号中断：无需计数，直接重试（非数据问题）
            continue;
        }
        else if (errno == EAGAIN || errno == EWOULDBLOCK)
        {
            // 暂时无数据：休眠后重试，累计次数
            retry_count++;
            usleep(SLEEP_US); // 休眠指定时间（微秒）
            if (retry_count >= MAX_RETRIES)
            {
                // LOG("Timeout: max retries (%d) send_nolock\n", MAX_RETRIES);

                return -1;
            }
        }
        else
        {
            // 其他错误（如文件描述符无效等）
            return -1;
        }
    }

    if (is_noblock == 0)
    {
        // 如果之前是阻塞的设置回去
        set_socket_blocking(fd);
    }

    return total;
}

int send_file_data(int sockfd, char *filename, off_t start_offset, size_t send_file_size)
{
    LOG("send_file_data: %s\n", filename);
    int file_fd = open(filename, O_RDONLY);
    if (file_fd == -1)
    {
        perror("master: open file failed");
        return -1;
    }

    off_t ret = lseek(file_fd, start_offset, SEEK_SET);
    if (ret == -1)
    {
        perror("lseek failed");
        close(file_fd);
        return -1;
    }

    off_t total_send = 0;
    ssize_t sent_len;
    while (total_send < send_file_size)
    {
        // LOG("total_len: %ld, send_file_size: %ld\n", total_send, send_file_size);
        sent_len = sendfile(sockfd, file_fd, NULL, send_file_size - total_send);
        if (sent_len == -1)
        {
            if (errno == EINTR)
                continue;
            perror("master: sendfile failed");
            close(file_fd);
            return -1;
        }
        total_send += sent_len;
    }
    close(file_fd);

    return total_send; // 返回发送的数据大小
}

/**
 * 快速置换两颗红黑树的内容（O(1)时间复杂度）
 * @param tree_a 第一颗红黑树（需初始化完成，非NULL）
 * @param tree_b 第二颗红黑树（需初始化完成，非NULL）
 * @return 0成功，-1失败（入参无效）
 */
int rb_swap(RBTree *tree_a, RBTree *tree_b)
{
    // 1. 入参校验（两颗树都必须存在且初始化完成）
    if (!tree_a || !tree_b)
    {
        // fprintf(stderr, "rb_swap: invalid tree pointer (NULL)\n");
        return -1;
    }
    if (tree_a->NIL_NODE == NULL || tree_b->NIL_NODE == NULL)
    {
        fprintf(stderr, "rb_swap: tree not initialized (NIL_NODE is NULL)\n");
        return -1;
    }

    // 2. 给两颗树加写锁（必须写锁，因为修改核心结构）
#if RB_TREE_MUTEX_LOCK
    // 互斥锁：加锁顺序无关（都是写锁，避免死锁只需确保两颗树都加锁）
    pthread_mutex_lock(&tree_a->mutex_lock);
    pthread_mutex_lock(&tree_b->mutex_lock);
#elif RB_TREE_RW_LOCK
    // 读写锁：加写锁（修改核心数据，读锁不足以保护）
    pthread_rwlock_wrlock(&tree_a->rw_lock);
    pthread_rwlock_wrlock(&tree_b->rw_lock);
#endif

    // 3. 保存tree_a的核心成员到临时变量
    Node *tmp_root = tree_a->root;
    int tmp_size = tree_a->size;
    Node *tmp_nil = tree_a->NIL_NODE;

    // 4. 交换核心成员：tree_a <- tree_b
    tree_a->root = tree_b->root;
    tree_a->size = tree_b->size;
    tree_a->NIL_NODE = tree_b->NIL_NODE;

    // 5. 交换核心成员：tree_b <- 临时保存的tree_a数据
    tree_b->root = tmp_root;
    tree_b->size = tmp_size;
    tree_b->NIL_NODE = tmp_nil;

    // 6. 解锁（顺序与加锁相反，规范操作）
#if RB_TREE_MUTEX_LOCK
    pthread_mutex_unlock(&tree_b->mutex_lock);
    pthread_mutex_unlock(&tree_a->mutex_lock);
#elif RB_TREE_RW_LOCK
    pthread_rwlock_unlock(&tree_b->rw_lock);
    pthread_rwlock_unlock(&tree_a->rw_lock);
#endif

    printf("rb_swap: two red-black trees swapped successfully (size_a: %d, size_b: %d)\n",
           tree_a->size, tree_b->size);
    return 0;
}

/**
 * 快速置换两颗哈希表的内容（O(1)时间复杂度）
 * @param ht_a 第一颗哈希表（需初始化完成，非NULL）
 * @param ht_b 第二颗哈希表（需初始化完成，非NULL）
 * @return 0成功，-1失败（入参无效/桶数量不一致）
 */
int hashtable_swap(HashTable *ht_a, HashTable *ht_b)
{
    // 1. 入参校验
    if (!ht_a || !ht_b)
    {
        // fprintf(stderr, "hashtable_swap: invalid hash table pointer (NULL)\n");
        return -1;
    }
    if (!ht_a->buckets || !ht_b->buckets)
    {
        fprintf(stderr, "hashtable_swap: hash table not initialized (buckets is NULL)\n");
        return -1;
    }
    // 关键校验：两颗表的桶数量必须一致（否则交换后访问buckets会越界）
    if (ht_a->bucket_count != ht_b->bucket_count)
    {
        fprintf(stderr, "hashtable_swap: bucket count mismatch (a: %zu, b: %zu)\n",
                ht_a->bucket_count, ht_b->bucket_count);
        return -1;
    }

    // 2. 给两颗表加写锁（修改核心数据，必须写锁）
#if HASH_TABLE_MUTEX_LOCK
    pthread_mutex_lock(&ht_a->mutex_lock);
    pthread_mutex_lock(&ht_b->mutex_lock);
#elif HASH_TABLE_RW_LOCK
    pthread_rwlock_wrlock(&ht_a->rw_lock);
    pthread_rwlock_wrlock(&ht_b->rw_lock);
#endif

    // 3. 临时保存ht_a的核心成员
    HashNode **tmp_buckets = ht_a->buckets;
    size_t tmp_count = ht_a->count;

    // 4. 交换核心成员
    ht_a->buckets = ht_b->buckets;
    ht_a->count = ht_b->count;

    ht_b->buckets = tmp_buckets;
    ht_b->count = tmp_count;

    // 5. 解锁（顺序与加锁相反）
#if HASH_TABLE_MUTEX_LOCK
    pthread_mutex_unlock(&ht_b->mutex_lock);
    pthread_mutex_unlock(&ht_a->mutex_lock);
#elif HASH_TABLE_RW_LOCK
    pthread_rwlock_unlock(&ht_b->rw_lock);
    pthread_rwlock_unlock(&ht_a->rw_lock);
#endif

    printf("hashtable_swap: two hash tables swapped successfully (count_a: %zu, count_b: %zu)\n",
           ht_a->count, ht_b->count);
    return 0;
}

/**
 * 快速置换两颗跳表的内容（O(1)时间复杂度）
 * @param sl_a 第一颗跳表（需初始化完成，非NULL）
 * @param sl_b 第二颗跳表（需初始化完成，非NULL）
 * @return 0成功，-1失败（入参无效）
 */
int skiplist_swap(SkipList *sl_a, SkipList *sl_b)
{
    // 1. 入参校验
    if (!sl_a || !sl_b)
    {
        // fprintf(stderr, "skiplist_swap: invalid skiplist pointer (NULL)\n");
        return -1;
    }
    if (!sl_a->header || !sl_b->header)
    {
        fprintf(stderr, "skiplist_swap: skiplist not initialized (header is NULL)\n");
        return -1;
    }

    // 2. 给两颗跳表加写锁（修改核心数据，必须写锁）
#if SKIPLIST_MUTEX_LOCK
    pthread_mutex_lock(&sl_a->mutex_lock);
    pthread_mutex_lock(&sl_b->mutex_lock);
#elif SKIPLIST_RW_LOCK
    pthread_rwlock_wrlock(&sl_a->rw_lock);
    pthread_rwlock_wrlock(&sl_b->rw_lock);
#endif

    // 3. 临时保存sl_a的核心成员
    SNode *tmp_header = sl_a->header;
    int tmp_level = sl_a->level;
    int tmp_size = sl_a->size;

    // 4. 交换核心成员
    sl_a->header = sl_b->header;
    sl_a->level = sl_b->level;
    sl_a->size = sl_b->size;

    sl_b->header = tmp_header;
    sl_b->level = tmp_level;
    sl_b->size = tmp_size;

    // 5. 解锁（顺序与加锁相反）
#if SKIPLIST_MUTEX_LOCK
    pthread_mutex_unlock(&sl_b->mutex_lock);
    pthread_mutex_unlock(&sl_a->mutex_lock);
#elif SKIPLIST_RW_LOCK
    pthread_rwlock_unlock(&sl_b->rw_lock);
    pthread_rwlock_unlock(&sl_a->rw_lock);
#endif

    printf("skiplist_swap: two skiplists swapped successfully (size_a: %d, level_a: %d)\n",
           sl_a->size, sl_a->level);
    return 0;
}

int destroy_engines(RBTree *rb_tree, HashTable *hash_table, SkipList *skiplist)
{

    // 删除红黑树内容
    if (rb_tree)
    {
        rb_destroy(rb_tree);
    }

    if (hash_table)
    {
        hashtable_destroy(hash_table);
    }

    if (skiplist)
    {
        skiplist_destroy(skiplist);
    }

    return 0;
}

// ====================== 主机逻辑（长连接+实时推送） ======================
static void *master_handle_slave(void *arg)
{
    int slave_fd = (intptr_t)arg;
    printf("master: new slave connection (fd=%d)\n", slave_fd);

    // 1. 接收从机同步请求
    SlaveSyncReq req;
    ssize_t n = recv_nolock(slave_fd, &req, sizeof(req), 0);
    if (n != sizeof(req))
    {
        fprintf(stderr, "master: read sync req failed (fd=%d)\n", slave_fd);
        close(slave_fd);
        return NULL;
    }
    req.aof_offset = ntohll(req.aof_offset);

    if (strcmp(req.slave_id, SLAVE_ID) != 0)
    {
        // slave_id不匹配，拒绝连接
        close(slave_fd);
    }

    // 2. 计算同步策略
    MasterSyncResp resp = {0};

    off_t current_aof_size = atomic_load(&aof_file_sync_length);

    printf("master aof size:%ld, slave aof size:%ld\n", current_aof_size, req.aof_offset);
    printf("master id:%s, slave_master_id: %s\n", g_ms_config.master_id, req.master_id);
    if (strlen(req.master_id) == 0 || strcmp(req.master_id, g_ms_config.master_id) != 0)
    {
        resp.need_full_sync = 1;
        resp.rb_full_size = get_file_size(RB_FULL_FILE_NAME);
        resp.hash_full_size = get_file_size(HASH_FULL_FILE_NAME);
        resp.skiplist_full_size = get_file_size(SKIPLIST_FULL_FILE_NALE);

        resp.aof_start = 0;
        resp.aof_size = current_aof_size;
    }
    else if (req.aof_offset < current_aof_size)
    {
        resp.need_full_sync = 0;
        resp.aof_start = req.aof_offset;
        resp.aof_size = current_aof_size - req.aof_offset;
    }
    else
    {
        resp.need_full_sync = 0;
        resp.aof_start = current_aof_size;
        resp.aof_size = 0;
    }

    // 3. 发送同步响应（转换为网络序）
    MasterSyncResp resp_net = {
        .need_full_sync = resp.need_full_sync,
        .rb_full_size = htonll(resp.rb_full_size),
        .hash_full_size = htonll(resp.hash_full_size),
        .skiplist_full_size = htonll(resp.skiplist_full_size),
        .aof_size = htonll(resp.aof_size),
        .aof_start = htonll(resp.aof_start)};

    strncpy(resp_net.master_id, g_ms_config.master_id, MASTER_ID_LEN);
    resp_net.master_id[MASTER_ID_LEN] = '\0'; // 确保结束符

    if (send_nolock(slave_fd, &resp_net, sizeof(resp_net), MSG_NOSIGNAL) != sizeof(resp_net))
    {
        perror("master: send sync resp failed");
        close(slave_fd);
        return NULL;
    }

    // 4. 发送全量文件（若需要）
    if (resp.need_full_sync)
    {

        // 发送红黑树全量文件
        int send_len = send_file_data(slave_fd, RB_FULL_FILE_NAME, 0, resp.rb_full_size);
        if (send_len != resp.rb_full_size)
        {
            printf("rb_full_file send fail!\n");
            close(slave_fd);
            return NULL;
        }

        // 发送哈希表全量文件
        send_len = send_file_data(slave_fd, HASH_FULL_FILE_NAME, 0, resp.hash_full_size);
        if (send_len != resp.hash_full_size)
        {
            printf("hash_full_file send fail!\n");
            close(slave_fd);
            return NULL;
        }

        // 发送跳表全量文件
        send_len = send_file_data(slave_fd, SKIPLIST_FULL_FILE_NALE, 0, resp.skiplist_full_size);
        if (send_len != resp.skiplist_full_size)
        {
            printf("skiplist_full_file send fail!\n");
            close(slave_fd);
            return NULL;
        }

        printf("full_file_send success(rb: %ld bytes, hash: %ld, skiplist: %ld)\n", resp.rb_full_size, resp.hash_full_size, resp.skiplist_full_size);
    }

    // 5. 发送初始AOF增量（若需要）
    if (resp.aof_size > 0)
    {
        // 发送AOF增量大小
        uint64_t aof_size_net = htonll(resp.aof_size);
        if (send_nolock(slave_fd, &aof_size_net, sizeof(aof_size_net), MSG_NOSIGNAL) != sizeof(aof_size_net))
        {
            perror("master: send AOF size failed");
            close(slave_fd);
            return NULL;
        }

        // sendfile发送AOF增量
        int send_len = send_file_data(slave_fd, AOF_FILE_NAME, resp.aof_start, resp.aof_size);

        if (send_len != resp.aof_size)
        {
            fprintf(stderr, "master: AOF file send fail\n");
            close(slave_fd);
            return NULL;
        }
        printf("master: initial AOF sent (size=%ld, fd=%d)\n", resp.aof_size, slave_fd);
    }

    // 6. 长连接循环：实时监控AOF变化并推送增量
    printf("master: entering long connection loop (fd=%d)\n", slave_fd);
    off_t last_aof_size = resp.aof_start + resp.aof_size;

    int retry_num = 0;
    while (atomic_load(&g_ms_config.master_running))
    {
        // 检查AOF是否有新增内容
        off_t current_aof = atomic_load(&aof_file_sync_length);

        if (current_aof > last_aof_size)
        {
            off_t aof_increment = current_aof - last_aof_size;
            printf("master: AOF file_size increase (increase_size=%ld, fd=%d)\n", aof_increment, slave_fd);

            // 发送增量大小（网络序）
            uint64_t inc_net = htonll(aof_increment);
            if (send_nolock(slave_fd, &inc_net, sizeof(inc_net), MSG_NOSIGNAL) != sizeof(inc_net))
            {
                perror("master: send AOF increase size failed");
                break;
            }

            int send_len = send_file_data(slave_fd, AOF_FILE_NAME, last_aof_size, aof_increment);
            if (send_len != aof_increment)
            {
                fprintf(stderr, "master: increment send incomplete (fd=%d)\n", slave_fd);
                break;
            }
            last_aof_size = current_aof;
            printf("master: increment sent (total=%ld, fd=%d)\n", last_aof_size, slave_fd);
            retry_num = 0;
        }
        else
        {
            retry_num++;
            if (retry_num == 5)
            {
                // 等效发送探测包：防止僵尸连接，触发EPIPE得以断开！
                uint64_t inc_net = 0;
                if (send_nolock(slave_fd, &inc_net, sizeof(inc_net), MSG_NOSIGNAL) != sizeof(inc_net))
                {
                    perror("master: send AOF increase size failed");
                    break;
                }
                retry_num = 0;
            }
        }

        // 休眠1秒再检查
        sleep(1);
    }

    // 7. 连接关闭清理
    printf("master: slave connection closed (fd=%d)\n", slave_fd);
    close(slave_fd);
    return NULL;
}

static void *master_listen_thread(void *arg)
{
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd == -1)
    {
        perror("master: create socket failed");
        return NULL;
    }

    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_addr.s_addr = INADDR_ANY,
        .sin_port = htons(g_ms_config.master_listen_port)};

    if (bind(listen_fd, (struct sockaddr *)&addr, sizeof(addr)) == -1)
    {
        perror("master: bind failed");
        close(listen_fd);
        return NULL;
    }

    if (listen(listen_fd, 10) == -1)
    {
        perror("master: listen failed");
        close(listen_fd);
        return NULL;
    }

    printf("master: listening on port %d (MasterID: %s)\n",
           g_ms_config.master_listen_port, g_ms_config.master_id);

    while (atomic_load(&g_ms_config.master_running))
    {
        struct sockaddr_in slave_addr;
        socklen_t addr_len = sizeof(slave_addr);
        int slave_fd = accept(listen_fd, (struct sockaddr *)&slave_addr, &addr_len);
        if (slave_fd == -1)
        {
            if (errno == EINTR)
                continue;
            perror("master: accept failed");
            break;
        }

        printf("master: slave connected %s:%d (fd=%d)\n",
               inet_ntoa(slave_addr.sin_addr), ntohs(slave_addr.sin_port), slave_fd);

        // 为每个从机连接创建独立线程
        pthread_t handle_thread;
        if (pthread_create(&handle_thread, NULL, master_handle_slave, (void *)(intptr_t)slave_fd) != 0)
        {
            perror("master: create thread failed");
            close(slave_fd);
        }
        pthread_detach(handle_thread);
    }

    close(listen_fd);
    printf("master: listen thread exited\n");
    return NULL;
}

// ====================== 从机逻辑（长连接+持续接收） ======================
int slave_splice_recv_with_offset(int sock_fd, const char *filename, off_t file_size, off_t offset)
{
    int fd = open(filename, O_WRONLY | O_CREAT, 0644);
    if (fd == -1)
    {
        perror("slave_splice_recv_with_offset: open failed");
        return -1;
    }

    if (offset > 0 && lseek(fd, offset, SEEK_SET) == -1)
    {
        perror("slave_splice_recv_with_offset: lseek failed");
        close(fd);
        return -1;
    }

    int pipe_fd[2];
    if (pipe(pipe_fd) == -1)
    {
        perror("slave_splice_recv_with_offset: pipe create failed");
        close(fd);
        return -1;
    }

    fcntl(pipe_fd[0], F_SETFL, fcntl(pipe_fd[0], F_GETFL) | O_NONBLOCK);
    fcntl(pipe_fd[1], F_SETFL, fcntl(pipe_fd[1], F_GETFL) | O_NONBLOCK);

    off_t total_received = 0;
    while (total_received < file_size)
    {
        size_t len = file_size - total_received;
        if (len > 1024 * 1024)
            len = 1024 * 1024;

        ssize_t n = splice(
            sock_fd, NULL,
            pipe_fd[1], NULL,
            len,
            SPLICE_F_MOVE | SPLICE_F_NONBLOCK);

        if (n <= 0)
        {
            if (n == 0)
                break;
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                usleep(1000);
                continue;
            }
            perror("slave_splice_recv_with_offset: splice socket to pipe failed");
            close(pipe_fd[0]);
            close(pipe_fd[1]);
            close(fd);
            return -1;
        }

        ssize_t m = splice(
            pipe_fd[0], NULL,
            fd, NULL,
            n,
            SPLICE_F_MOVE | SPLICE_F_NONBLOCK);

        if (m <= 0)
        {
            if (m == 0)
                break;
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                usleep(1000);
                continue;
            }
            perror("slave_splice_recv_with_offset: splice pipe to file failed");
            close(pipe_fd[0]);
            close(pipe_fd[1]);
            close(fd);
            return -1;
        }

        total_received += m;
        // printf("slave: received %ld/%ld bytes (offset=%ld)\n", total_received, file_size, offset);
    }

    close(pipe_fd[0]);
    close(pipe_fd[1]);
    fsync(fd);
    close(fd);

    return (total_received == file_size) ? 0 : -1;
}

static void *slave_sync_thread(void *arg)
{
    g_ms_config.slave_meta = read_slave_meta();
    printf("slave: initial meta - MasterID: %s, AOF Offset: %ld\n",
           g_ms_config.slave_meta.master_id[0] ? g_ms_config.slave_meta.master_id : "N/A",
           g_ms_config.slave_meta.aof_offset);

    // 建立长连接
    int sock_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (sock_fd == -1)
    {
        perror("slave: create socket failed");
        return NULL;
    }

    struct sockaddr_in master_addr = {
        .sin_family = AF_INET,
        .sin_addr.s_addr = inet_addr(g_ms_config.master_ip),
        .sin_port = htons(g_ms_config.master_port)};

    if (connect(sock_fd, (struct sockaddr *)&master_addr, sizeof(master_addr)) == -1)
    {
        perror("slave: connect failed");
        close(sock_fd);
        return NULL;
    }
    printf("slave: connected to master %s:%d (fd=%d)\n",
           g_ms_config.master_ip, g_ms_config.master_port, sock_fd);

    // 发送同步请求（网络序转换）
    SlaveSyncReq req = {
        //.slave_id = 0,
        .aof_offset = htonll(g_ms_config.slave_meta.aof_offset)};
    snprintf(req.slave_id, SLAVE_ID_MAX_LEN + 1, "%s", g_ms_config.slave_id);
    strncpy(req.master_id, g_ms_config.slave_meta.master_id, MASTER_ID_LEN);
    req.master_id[MASTER_ID_LEN] = '\0'; // 确保结束符
    if (send_nolock(sock_fd, &req, sizeof(req), MSG_NOSIGNAL) != sizeof(req))
    {
        perror("slave: send sync req failed");
        close(sock_fd);
        return NULL;
    }

    // 接收主机响应（转换为本地序）
    MasterSyncResp resp_net;
    ssize_t n = recv_nolock(sock_fd, &resp_net, sizeof(resp_net), 0);
    if (n != sizeof(resp_net))
    {
        perror("slave: read sync resp failed");
        close(sock_fd);
        return NULL;
    }
    MasterSyncResp resp = {
        .need_full_sync = resp_net.need_full_sync,
        .rb_full_size = ntohll(resp_net.rb_full_size),
        .hash_full_size = ntohll(resp_net.hash_full_size),
        .skiplist_full_size = ntohll(resp_net.skiplist_full_size),

        .aof_size = ntohll(resp_net.aof_size),
        .aof_start = ntohll(resp_net.aof_start),

        //.master_id = resp_net.master_id  // 注意不能直接这样赋值，数组是常量指针！！！
    };
    strncpy(resp.master_id, resp_net.master_id, MASTER_ID_LEN);
    resp.master_id[MASTER_ID_LEN] = '\0';

    // ==================初始化同步备份数据结构==================
    RBTree *sync_backup_rbtree = NULL;
    HashTable *sync_backup_hash_table = NULL;
    SkipList *sync_backup_skiplist = NULL;

    if (g_ms_config.slave_rbtree)
    {
        sync_backup_rbtree = rb_init();
        if (!sync_backup_rbtree)
        {
            fprintf(stderr, "slave: sync_backup_rbtree init failed\n");
            close(sock_fd);
            return NULL;
        }
    }

    if (g_ms_config.slave_hash_table)
    {
        sync_backup_hash_table = hashtable_create(g_ms_config.slave_hash_table->bucket_count);
        if (!sync_backup_hash_table)
        {
            fprintf(stderr, "slave: sync_backup_hash_table init failed\n");
            close(sock_fd);
            return NULL;
        }
    }

    if (g_ms_config.slave_skiplist)
    {
        sync_backup_skiplist = skiplist_init();
        if (!sync_backup_skiplist)
        {
            fprintf(stderr, "slave: sync_backup_skiplist init failed\n");
            close(sock_fd);
            return NULL;
        }
    }

    // 处理全量同步
    if (resp.need_full_sync)
    {
        clear_file(AOF_FILE_NAME);
        printf("清空AOF 文件\n");
        clear_file(RB_FULL_FILE_NAME);
        clear_file(HASH_FULL_FILE_NAME);
        clear_file(SKIPLIST_FULL_FILE_NALE);
        printf("清空FULL 文件\n");
        printf("slave: need full sync (rb_full_size = %ld, hash_full_size = %ld, skiplist_full_size = %ld)\n", resp.rb_full_size, resp.hash_full_size, resp.skiplist_full_size);

        // 接收红黑树全量文件
        if (slave_splice_recv_with_offset(sock_fd, RB_FULL_FILE_NAME, resp.rb_full_size, 0) != 0)
        {
            fprintf(stderr, "slave: receive rb_full failed\n");
            goto cleanup;
        }
        printf("slave: recv rb_full_file success\n");

        // 接收哈希表全量文件
        if (slave_splice_recv_with_offset(sock_fd, HASH_FULL_FILE_NAME, resp.hash_full_size, 0) != 0)
        {
            fprintf(stderr, "slave: receive rb_full failed\n");
            goto cleanup;
        }
        printf("slave: recv hash_full_file success\n");

        // 接收跳表全量文件
        if (slave_splice_recv_with_offset(sock_fd, SKIPLIST_FULL_FILE_NALE, resp.skiplist_full_size, 0) != 0)
        {
            fprintf(stderr, "slave: receive rb_full failed\n");
            goto cleanup;
        }
        printf("slave: recv skiplist_full_file success\n");

        if (g_ms_config.slave_rbtree)
        {
            rb_full_restore(RB_FULL_FILE_NAME, sync_backup_rbtree);
            rb_swap(g_ms_config.slave_rbtree, sync_backup_rbtree);
#if SLAVE_START_SYNC_BACKUP_ENGINES
            rb_full_restore(RB_FULL_FILE_NAME, sync_backup_rbtree);
#endif
            printf("slave: slave_rbtree restore success\n");
        }
        if (g_ms_config.slave_hash_table)
        {
            hashtable_full_restore(HASH_FULL_FILE_NAME, sync_backup_hash_table);
            hashtable_swap(g_ms_config.slave_hash_table, sync_backup_hash_table);
#if SLAVE_START_SYNC_BACKUP_ENGINES
            hashtable_full_restore(HASH_FULL_FILE_NAME, sync_backup_hash_table);
#endif
            printf("slave: slave_hash restore success\n");
        }
        if (g_ms_config.slave_skiplist)
        {
            skiplist_full_restore(SKIPLIST_FULL_FILE_NALE, sync_backup_skiplist);
            skiplist_swap(g_ms_config.slave_skiplist, sync_backup_skiplist);
#if SLAVE_START_SYNC_BACKUP_ENGINES
            skiplist_full_restore(SKIPLIST_FULL_FILE_NALE, sync_backup_skiplist);
#endif
            printf("slave: slave_skiplist restore success\n");
        }

        // 全量同步成功后再保存master_id
        // 更新从机元数据中的主机ID
        strncpy(g_ms_config.slave_meta.master_id, resp.master_id, MASTER_ID_LEN);
        g_ms_config.slave_meta.master_id[MASTER_ID_LEN] = '\0';
        // 立即写入本地文件持久化
        write_slave_meta(&g_ms_config.slave_meta);
        printf("slave: saved master_id: %s\n", g_ms_config.slave_meta.master_id);
    }

    // 处理初始AOF增量
    off_t current_aof_offset = resp.aof_start;
    if (resp.aof_size > 0)
    {
        printf("slave: need initial AOF sync (size=%ld, start=%ld)\n", resp.aof_size, resp.aof_start);

        // 接收AOF大小确认
        uint64_t aof_size_net;
        if (recv_nolock(sock_fd, &aof_size_net, sizeof(aof_size_net), 0) != sizeof(aof_size_net))
        {
            perror("slave: read AOF size failed");
            // close(sock_fd);
            // return NULL;
            goto cleanup;
        }
        off_t aof_size = ntohll(aof_size_net);
        if (aof_size != resp.aof_size)
        {
            fprintf(stderr, "slave: AOF size mismatch (recv=%ld, expected=%ld)\n", aof_size, resp.aof_size);
            goto cleanup;
        }

        // 接收AOF增量
        if (slave_splice_recv_with_offset(sock_fd, AOF_FILE_NAME, aof_size, resp.aof_start) != 0)
        {
            fprintf(stderr, "slave: receive initial AOF failed\n");
            goto cleanup;
        }

        // 更新元数据
        g_ms_config.slave_meta.aof_offset = current_aof_offset + aof_size;
        write_slave_meta(&g_ms_config.slave_meta);
#if SLAVE_START_SYNC_BACKUP_ENGINES
        // 重放AOF到备份数据结构
        aof_replay(AOF_FILE_NAME, current_aof_offset, sync_backup_rbtree, sync_backup_hash_table, sync_backup_skiplist);
        rb_swap(g_ms_config.slave_rbtree, sync_backup_rbtree);
        hashtable_swap(g_ms_config.slave_hash_table, sync_backup_hash_table);
        skiplist_swap(g_ms_config.slave_skiplist, sync_backup_skiplist);
        aof_replay(AOF_FILE_NAME, current_aof_offset, sync_backup_rbtree, sync_backup_hash_table, sync_backup_skiplist);
#else
        // 重放AOF
        aof_replay(AOF_FILE_NAME, current_aof_offset, g_ms_config.slave_rbtree, g_ms_config.slave_hash_table, g_ms_config.slave_skiplist);
#endif
        current_aof_offset += aof_size;
        printf("slave: initial AOF replayed (new offset=%ld)\n", current_aof_offset);
    }

    // 长连接循环：持续接收主机推送的增量
    printf("slave: entering long connection loop (fd=%d)\n", sock_fd);

    // set_socket_noblock(sock_fd);
    while (atomic_load(&g_ms_config.slave_running))
    {

        // 读取主机推送的增量大小
        uint64_t inc_net;
        ssize_t n = recv_nolock(sock_fd, &inc_net, sizeof(inc_net), 0);
        if (n < 0)
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                usleep(1000); // 数据暂不可用，休眠1ms后重试
                continue;
            }
            perror("slave: read increment size failed");
            break;
        }
        else if (n == 0)
        {
            printf("slave: master closed connection (fd=%d)\n", sock_fd);
            break;
        }

        off_t aof_increment = ntohll(inc_net);
        if (aof_increment < 0)
        {
            fprintf(stderr, "slave: invalid increment size %ld\n", aof_increment);
            break;
        }
        else if (aof_increment == 0)
        {
            usleep(100000);
            continue;
        }

        // 接收增量内容
        if (slave_splice_recv_with_offset(sock_fd, AOF_FILE_NAME, aof_increment, current_aof_offset) != 0)
        {
            fprintf(stderr, "slave: receive increment failed\n");
            break;
        }
        printf("slave: receiving AOF increment (size=%ld, offset=%ld)\n", aof_increment, current_aof_offset);

        // 实时更新元数据，防止断电没记录下来
        g_ms_config.slave_meta.aof_offset = current_aof_offset + aof_increment;
        write_slave_meta(&g_ms_config.slave_meta);
        // printf("slave: increment replayed (new offset=%ld)\n", current_aof_offset);
#if SLAVE_START_SYNC_BACKUP_ENGINES
        // 重放AOF到备份数据结构
        aof_replay(AOF_FILE_NAME,  current_aof_offset, sync_backup_rbtree, sync_backup_hash_table, sync_backup_skiplist);
        rb_swap(g_ms_config.slave_rbtree, sync_backup_rbtree);
        hashtable_swap(g_ms_config.slave_hash_table, sync_backup_hash_table);
        skiplist_swap(g_ms_config.slave_skiplist, sync_backup_skiplist);
        aof_replay(AOF_FILE_NAME, current_aof_offset, sync_backup_rbtree, sync_backup_hash_table, sync_backup_skiplist);
#else
        // 重放AOF
        aof_replay(AOF_FILE_NAME, current_aof_offset, g_ms_config.slave_rbtree, g_ms_config.slave_hash_table, g_ms_config.slave_skiplist);
#endif

        current_aof_offset += aof_increment;
    }

    destroy_engines(sync_backup_rbtree, sync_backup_hash_table, sync_backup_skiplist);

    // 连接关闭清理
    printf("slave: connection closed (fd=%d)\n", sock_fd);
    close(sock_fd);
    return NULL;

cleanup:
    destroy_engines(sync_backup_rbtree, sync_backup_hash_table, sync_backup_skiplist);
    printf("slave: connection closed (fd=%d)\n", sock_fd);
    close(sock_fd);
    return NULL;
}

// ====================== 公共接口（初始化随机种子） ======================
int ms_init(ServerMode mode, const char *master_ip, int master_port, int listen_port, RBTree *rbtree, HashTable *hash_table, SkipList *skiplist)
{
    memset(&g_ms_config, 0, sizeof(MasterSlaveConfig));
    g_ms_config.mode = mode;
    atomic_init(&g_ms_config.master_running, false);
    atomic_init(&g_ms_config.slave_running, false);

    // 初始化随机种子（仅主机需要，从机可选）
    srand((unsigned int)time(NULL) ^ (unsigned long)pthread_self());

    switch (mode)
    {
    case MODE_MASTER:
        // 引擎配置
        g_ms_config.master_rbtree = rbtree;
        g_ms_config.master_hash_table = hash_table;
        g_ms_config.master_skiplist = skiplist;

        // 主机端口配置
        g_ms_config.master_listen_port = listen_port;

        // id生成
        generate_master_id(g_ms_config.master_id); // 用rand()生成主机ID

        // 可信任的从机id
        snprintf(g_ms_config.slave_id, SLAVE_ID_MAX_LEN + 1, SLAVE_ID);

        break;

    case MODE_SLAVE:
        // 引擎配置
        g_ms_config.slave_rbtree = rbtree;
        g_ms_config.slave_hash_table = hash_table;
        g_ms_config.slave_skiplist = skiplist;

        // ip、端口配置
        strncpy(g_ms_config.master_ip, master_ip, sizeof(g_ms_config.master_ip) - 1);
        g_ms_config.master_port = master_port;

        // 历史记录读取
        g_ms_config.slave_meta = read_slave_meta();

        // 发送给主机验证的从机id
        snprintf(g_ms_config.slave_id, SLAVE_ID_MAX_LEN + 1, SLAVE_ID);

        break;

    case MODE_STANDALONE:
        return 0;
    default:
        return -1;
    }
    return 0;
}

int ms_start()
{
    switch (g_ms_config.mode)
    {
    case MODE_MASTER:
        atomic_store(&g_ms_config.master_running, true);
        if (pthread_create(&g_ms_config.master_listen_thread, NULL, master_listen_thread, NULL) != 0)
        {
            perror("ms_start: create master thread failed");
            return -1;
        }
        break;
    case MODE_SLAVE:
        atomic_store(&g_ms_config.slave_running, true);
        if (pthread_create(&g_ms_config.slave_sync_thread, NULL, slave_sync_thread, NULL) != 0)
        {
            perror("ms_start: create slave thread failed");
            return -1;
        }
        break;
    default:
        break;
    }
    return 0;
}

void ms_stop()
{
    switch (g_ms_config.mode)
    {
    case MODE_MASTER:
        atomic_store(&g_ms_config.master_running, false);
        pthread_kill(g_ms_config.master_listen_thread, SIGINT);
        pthread_join(g_ms_config.master_listen_thread, NULL);
        break;
    case MODE_SLAVE:
        atomic_store(&g_ms_config.slave_running, false);
        pthread_join(g_ms_config.slave_sync_thread, NULL);
        break;
    default:
        break;
    }
}

#endif